package profiler;

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;

import dataStructures.ProfileWritable;

public class SplitOutputReducer extends Reducer<Text, ProfileWritable, NullWritable, ProfileWritable> {

	private MultipleOutputs<NullWritable, ProfileWritable> mos;
	
	@Override
	protected void cleanup(Context context) throws IOException, InterruptedException {
		super.cleanup(context);
		mos.close();
	}


	@Override
	protected void setup(Context context) throws IOException, InterruptedException {
		super.setup(context);
		mos = new MultipleOutputs<NullWritable, ProfileWritable>(context);
	}


	@Override
	protected void reduce(Text key, Iterable<ProfileWritable> value, Context context) throws IOException, InterruptedException {
		 
		NullWritable n = NullWritable.get();
		for (ProfileWritable p : value) {
			
			context.write(n, p);
		}
		}
	

}
